package com.bigdata.assignment.problem2;

import java.io.IOException;
import java.text.Normalizer;
import java.util.regex.Pattern;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

/**
 * WordCountMapper - 支持 Unicode，规范化去变音符
 */
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    private final static IntWritable one = new IntWritable(1);
    private final Text word = new Text();

    private static final Pattern SPLIT_PATTERN = Pattern.compile("[^\\p{L}\\p{N}]+");

    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {

        if (value == null) return;

        String line = value.toString().toLowerCase();
        String normalized = Normalizer.normalize(line, Normalizer.Form.NFD).replaceAll("\\p{M}", "");
        String[] tokens = SPLIT_PATTERN.split(normalized);

        for (String token : tokens) {
            if (token == null) continue;
            token = token.trim();
            if (token.isEmpty()) continue;
            word.set(token);
            context.write(word, one);
        }
    }
}
